// Copyright 2022 Huawei Cloud Computing Technology Co., Ltd.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "CaeChannelDataHandler.h"
#include "../cae_common/CaeLogConfig.h"
#include "CaeSocketAgent.h"
#include "../cae_agent/CaeAgent.h"

const uint16_t MSG_MAGICWORD = 0x5A5A;
const uint16_t HEARTBEAT_MSG_ID = 0;
const uint16_t DATA_MSG_ID = 1;
const int MSG_HEADER_LEN = 6;
const uint32_t DATA_MAX_LEN = 5120; //5K

CaeChannelDataRecvHandler::CaeChannelDataRecvHandler(CaeChannelConnection *connection)
        : mChannelConnection(connection) {}

CaeChannelDataRecvHandler::~CaeChannelDataRecvHandler()
{
    mChannelConnection = nullptr;
}

void CaeChannelDataRecvHandler::Run()
{
    char threadName[] = "ChannelRecvHandler";
    prctl(PR_SET_NAME, threadName);
    int ret = 0;
    uint8_t data[MSG_HEADER_LEN];
    CaeChannelDataHeader_t header;
    CaeChannelDataBody_t body;
    LOGI("Channel data recv handler thread run.");
    while (m_threadStatus == CAE_THREAD_STATUS_RUNNING) {
        ret = mChannelConnection->ReadN(data, MSG_HEADER_LEN);
        if (ret != MSG_HEADER_LEN) {
            if (ret < 0) {
                LOGE("Read msg header failed.");
                break;
            }
            usleep(100000);//100ms
            continue;
        } else {
            header.id = (data[0] << 8) | data[1];
            header.length = (data[2] << 8) | data[3];
            header.magicword = (data[4] << 8) | data[5];
        }
        if (header.id != HEARTBEAT_MSG_ID) {
            body.len = header.length;
            if (body.len > DATA_MAX_LEN) {
                LOGE("Invalid body len: %d", body.len);
                break;
            }
            body.data = new uint8_t[body.len];
            ret = mChannelConnection->ReadN(body.data, body.len);
            if (ret != static_cast<int>(body.len)) {
                LOGE("Read body msg failed, expect: %d, actual: %d", body.len, ret);
            }
        } else {
            body.len = 0;
            body.data = nullptr;
        }

        ret = HandleRequest(header, body);
        if (body.data != nullptr) {
            delete []body.data;
            body.data = nullptr;
        }
        if (ret < 0) {
            LOGE("Handle request error.");
            break;
        }
    }
    m_threadStatus = CAE_THREAD_STATUS_STOP;
    LOGI("Channel data recv handler thread exit.");
}

int CaeChannelDataRecvHandler::HandleRequest(CaeChannelDataHeader_t &header,
                                             CaeChannelDataBody_t &body)
{
    if (header.magicword != MSG_MAGICWORD) {
        LOGE("Invalid header magigword.");
        return -1;
    }
    if (header.id == HEARTBEAT_MSG_ID) {
        uint8_t data[MSG_HEADER_LEN] = {0};
        data[0] = 0;
        data[1] = 0;
        data[2] = 0;
        data[3] = 0;
        data[4] = (MSG_MAGICWORD & 0xFF00) >> 8;
        data[5] = MSG_MAGICWORD & 0x00FF;
        int ret = mChannelConnection->WriteN(data, MSG_HEADER_LEN);
        if (ret != MSG_HEADER_LEN) {
            LOGE("Send heartbeat response failed.");
        }
    } else {
        uint32_t size = sizeof(STREAM_MSG_HEAD) + body.len;
        std::shared_ptr<uint8_t> tmpBuff;
        tmpBuff = std::shared_ptr<uint8_t>(new uint8_t[size], std::default_delete<uint8_t[]>());
        if (tmpBuff == nullptr) {
            LOGE("New tmp buf failed.");
            m_threadStatus = CAE_THREAD_STATUS_STOP;
            return -1;
        }
        StreamMsgHead *msgHead = (StreamMsgHead *)tmpBuff.get();
        msgHead->type = CAE_MSG_CHANNEL;
        msgHead->SetPayloadSize(body.len);
        msgHead->magicWord = CAE_STREAM_DELIMITER_MAGICWORD;
        msgHead->checksum = (unsigned char) GET_CAE_CHECKSUM(CAE_MSG_CHANNEL);
        LOGI("Channel socket recv msgId: %d, msgLen: %d", header.id, header.length);
        if (EOK != memcpy_s(tmpBuff.get(), sizeof(STREAM_MSG_HEAD), msgHead, sizeof(STREAM_MSG_HEAD))) {
            LOGE("Mem copy header failed.");
        }
        if (EOK != memcpy_s(tmpBuff.get() + sizeof(STREAM_MSG_HEAD), body.len, body.data, body.len)) {
            LOGE("Mem copy body failed.");
        }

        int ret = CaeAgent::GetInstance().SendCustomDataToClient(tmpBuff, size, msgHead->type);
        LOGI("Send channel data to client, %d", ret);
    }
    return 0;
}

CaeChannelDataSendHandler::CaeChannelDataSendHandler(CaeChannelConnection *mChannelConnection)
        : mChannelConnection(mChannelConnection) {}

CaeChannelDataSendHandler::~CaeChannelDataSendHandler()
{
    mChannelConnection = nullptr;
}

void CaeChannelDataSendHandler::Run()
{
    char threadName[] = "ChannelSendHandler";
    prctl(PR_SET_NAME, threadName);
    CaeDataPipe *dataPipe = nullptr;
    for (auto pipe : CaeAgent::GetInstance().m_caePipes) {
        if (nullptr != pipe && pipe->m_type == CAE_MSG_CHANNEL) {
            dataPipe = pipe;
            break;
        }
    }
    if (dataPipe == nullptr) {
        LOGE("Channel data pipe is null.");
        m_threadStatus = CAE_THREAD_STATUS_STOP;
        return;
    }
    LOGI("Channel data send handler thread run.");
    while (m_threadStatus == CAE_THREAD_STATUS_RUNNING) {
        CaePipeData *pipeData = dataPipe->GetPipeData();
        if (pipeData == nullptr || pipeData->GetLen() <= 0) {
            usleep(100000); //100ms
            continue;
        }
        uint16_t msgLen = pipeData->GetLen() + MSG_HEADER_LEN;
        uint8_t *data = new(std::nothrow) uint8_t[msgLen];
        if (data == nullptr) {
            LOGE("New tmp buf failed.");
            break;
        }
        data[0] = DATA_MSG_ID >> 8;
        data[1] = DATA_MSG_ID & 0xFF;
        data[2] = (pipeData->GetLen() & 0xFF00) >> 8;
        data[3] = pipeData->GetLen() & 0x00FF;
        data[4] = (MSG_MAGICWORD & 0xFF00) >> 8;
        data[5] = MSG_MAGICWORD & 0x00ff;
        LOGI("Channel socket send msgId: %d, msgLen: %d", DATA_MSG_ID, msgLen - MSG_HEADER_LEN);
        if (EOK != memcpy_s(data + MSG_HEADER_LEN, msgLen, pipeData->GetBuffer(), pipeData->GetLen())) {
            LOGE("Mem copy pipe data failed.");
        } else {
            int ret = mChannelConnection->WriteN(data, msgLen);
            if (ret != msgLen) {
                LOGE("Write msg failed, expect: %d, actual: %d", msgLen, ret);
            }
        }
        delete []data;
        delete pipeData;
        pipeData = nullptr;
    }
    LOGI("Channel data send handler thread exit.");
}